import org.apache.spark.sql.SparkSession

object data1_core4 {
  def main(args: Array[String]): Unit = {
    val spark=SparkSession
      .builder
      .master("local[*]")
      .appName("Spark Pi")
      .getOrCreate()
    val sc =spark.sparkContext
    //键值对key&value一个键值对就是一个元组
    val data1=sc.makeRDD(List(("张三",2000),("李四",2500),("王五",5000),("张三",3600)))
    val data2=sc.makeRDD(List(("张三",3000),("赵四",3500),("王五",5000),("张三",4600)))
    //聚合(将相同key的值累加)
    data1.reduceByKey((x,y)=>x+y).foreach(println)
    //将相同key的值生成一个新的序列
    data1.groupByKey().foreach(println)
    //合并join相同的key逐一匹配，不同的key就不会出现
    data1.join(data2).foreach(println)
    //左（右）合并是以左（右）为主，相同的key逐一匹配，没有的key就为none
    data1.leftOuterJoin(data2).foreach(println)
    data1.rightOuterJoin(data2).foreach(println)
    //合并
    data1.combineByKey(
      v=>(v,1),
      (t:(Int,Int),v)=>{
        (t._1+v,t._2+1)
      },
      (t1: (Int, Int), t2:(Int,Int)) => {
        (t1._1 + t2._2, t1._2 + t2._2)
      }
    ).foreach(println)
    //聚会（求最大（小）值）
    data1.aggregateByKey(0)(math.max(_,_),_+_).foreach(println)
    sc.stop()
  }

}
